package com.example.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently {

    DefaultMQProducer replyProducer;

    @Resource
    public void setDefaultMQProducer(DefaultMQProducer replyProducer) {
        this.replyProducer = replyProducer;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        MessageExt messageExt = list.get(0);
        System.out.printf("%s order服务接受到消息: %s %n", Thread.currentThread().getName(), new String(messageExt.getBody()).toString());
        try {
            // 异步发送回调消息，需要发送消息给生产者，否则会超时导致异常。
//            JSONObject jsonObject = JSONObject.parseObject(new String(messageExt.getBody()));
//            String couponId = jsonObject.get("couponId").toString();
//            System.out.printf("%s 号id", couponId);
            Message replyMessage = MessageUtil.createReplyMessage(messageExt, messageExt.getBody());
            replyProducer.send(replyMessage, 10000);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }



}
